package org.shj.spark.core;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;

public class BroadCastValue {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("BroadCastValue");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		final int f = 3;
		final Broadcast<Integer> broadCastFactor = sc.broadcast(f);
		List<Integer> list = Arrays.asList(1,2,3,4,5);
		JavaRDD<Integer> listRDD = sc.parallelize(list);
		listRDD.map(new Function<Integer, Integer>(){
			private static final long serialVersionUID = 8523891276294451896L;

			@Override
			public Integer call(Integer num) throws Exception {
				//使用 f 的话，f 会被封装到task里面去，会被发送到远程的从结点去执行。
				//每个task都会有一个 f 变量, 如果 f 是一个比较占内存的变量，则建议使用broadcase
				//return num*f;
				return num * broadCastFactor.value(); //
			}
			
		}).foreach(new VoidFunction<Integer>(){
			private static final long serialVersionUID = -8616397941580884795L;

			@Override
			public void call(Integer t) throws Exception {
				System.out.println(t);
			}
			
		});;
		
		sc.close();

	}

}
